package org.apache.nifi.redis.service;

import cn.hutool.core.util.StrUtil;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnDisabled;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.redis.RedisDistributedLock;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author 酷酷的诚
 * @date 2019-08-22 15:41
 */
@Tags({"redis", "distributed", "lock"})
@CapabilityDescription("Redis缓存锁，只支持单节点Redis")
public class RedisDistributedLockService extends AbstractControllerService implements RedisDistributedLock {

    private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    private static final Long RELEASE_SUCCESS = 1L;

    public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder()
            .name("Connection String")
            .displayName("Connection String")
            .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " +
                    "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " +
                    "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.")
            .required(true)
            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder()
            .name("Database Index")
            .displayName("Database Index")
            .description("The database index to be used by connections created from this connection pool. " +
                    "See the databases property in redis.conf, by default databases 0-15 will be available.")
            .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
            .defaultValue("0")
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .required(true)
            .build();

    public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder()
            .name("Communication Timeout")
            .displayName("Communication Timeout")
            .description("The timeout to use when attempting to communicate with Redis.")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .defaultValue("10 seconds")
            .required(true)
            .build();
    public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
            .name("Password")
            .displayName("Password")
            .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.")
            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .sensitive(true)
            .build();

    public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder()
            .name("Pool - Max Total")
            .displayName("Pool - Max Total")
            .description("The maximum number of connections that can be allocated by the pool (checked out to clients, or idle awaiting checkout). " +
                    "A negative value indicates that there is no limit.")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .defaultValue("8")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder()
            .name("Pool - Max Idle")
            .displayName("Pool - Max Idle")
            .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .defaultValue("8")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder()
            .name("Pool - Min Idle")
            .displayName("Pool - Min Idle")
            .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " +
                    "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .defaultValue("0")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder()
            .name("Pool - Block When Exhausted")
            .displayName("Pool - Block When Exhausted")
            .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " +
                    "Setting this to false means an error will occur immediately when a client requests a connection and none are available.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("true")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder()
            .name("Pool - Max Wait Time")
            .displayName("Pool - Max Wait Time")
            .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .defaultValue("10 seconds")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder()
            .name("Pool - Min Evictable Idle Time")
            .displayName("Pool - Min Evictable Idle Time")
            .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .defaultValue("60 seconds")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder()
            .name("Pool - Time Between Eviction Runs")
            .displayName("Pool - Time Between Eviction Runs")
            .description("The amount of time between attempting to evict idle connections from the pool.")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .defaultValue("30 seconds")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder()
            .name("Pool - Num Tests Per Eviction Run")
            .displayName("Pool - Num Tests Per Eviction Run")
            .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .defaultValue("-1")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder()
            .name("Pool - Test On Create")
            .displayName("Pool - Test On Create")
            .description("Whether or not connections should be tested upon creation.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("false")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder()
            .name("Pool - Test On Borrow")
            .displayName("Pool - Test On Borrow")
            .description("Whether or not connections should be tested upon borrowing from the pool.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("false")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder()
            .name("Pool - Test On Return")
            .displayName("Pool - Test On Return")
            .description("Whether or not connections should be tested upon returning to the pool.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("false")
            .required(true)
            .build();

    public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder()
            .name("Pool - Test While Idle")
            .displayName("Pool - Test While Idle")
            .description("Whether or not connections should be tested while idle.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("true")
            .required(true)
            .build();
    public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS;

    static {
        final List<PropertyDescriptor> props = new ArrayList<>();
        props.add(CONNECTION_STRING);
        props.add(DATABASE);
        props.add(COMMUNICATION_TIMEOUT);
        props.add(PASSWORD);
        props.add(POOL_MAX_TOTAL);
        props.add(POOL_MAX_IDLE);
        props.add(POOL_MIN_IDLE);
        props.add(POOL_BLOCK_WHEN_EXHAUSTED);
        props.add(POOL_MAX_WAIT_TIME);
        props.add(POOL_MIN_EVICTABLE_IDLE_TIME);
        props.add(POOL_TIME_BETWEEN_EVICTION_RUNS);
        props.add(POOL_NUM_TESTS_PER_EVICTION_RUN);
        props.add(POOL_TEST_ON_CREATE);
        props.add(POOL_TEST_ON_BORROW);
        props.add(POOL_TEST_ON_RETURN);
        props.add(POOL_TEST_WHILE_IDLE);
        REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
    }

    private volatile PropertyContext context;
    private volatile JedisPool jedisPool;

    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return REDIS_CONNECTION_PROPERTY_DESCRIPTORS;
    }

    @Override
    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
        final List<ValidationResult> results = new ArrayList<>();

        final String connectionString = validationContext.getProperty(CONNECTION_STRING).evaluateAttributeExpressions().getValue();
        final Integer dbIndex = validationContext.getProperty(DATABASE).evaluateAttributeExpressions().asInteger();

        if (StrUtil.isBlank(connectionString)) {
            results.add(new ValidationResult.Builder()
                    .subject(CONNECTION_STRING.getDisplayName())
                    .valid(false)
                    .explanation("Connection String cannot be blank")
                    .build());
        } else {
            final String[] hostAndPort = connectionString.split("[:]");
            if (hostAndPort.length != 2 || StrUtil.isBlank(hostAndPort[0]) || StrUtil.isBlank(hostAndPort[1]) || !isInteger(hostAndPort[1])) {
                results.add(new ValidationResult.Builder()
                        .subject(CONNECTION_STRING.getDisplayName())
                        .input(connectionString)
                        .valid(false)
                        .explanation("Standalone Connection String must be in the form host:port")
                        .build());
            }
        }

        if (dbIndex > 0) {
            results.add(new ValidationResult.Builder()
                    .subject(DATABASE.getDisplayName())
                    .valid(false)
                    .explanation("Database Index must be 0 when using clustered Redis")
                    .build());
        }

        return results;

    }

    @OnEnabled
    public void onEnabled(final ConfigurationContext context) {
        this.context = context;
        this.jedisPool = getJedisPool();
    }

    @OnDisabled
    public void onDisabled() {
        if (jedisPool != null) {
            jedisPool.destroy();
            jedisPool = null;
            context = null;
        }
    }

    public JedisPool getJedisPool() {
        if (jedisPool == null) {
            synchronized (this) {
                if (jedisPool == null) {
                    jedisPool = createJedisPool();
                }
            }
        }
        return jedisPool;
    }


    private JedisPool createJedisPool() {
        final String connectionString = context.getProperty(CONNECTION_STRING).evaluateAttributeExpressions().getValue();
        final Integer dbIndex = context.getProperty(DATABASE).evaluateAttributeExpressions().asInteger();
        final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
        final int timeout = context.getProperty(COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
        final JedisPoolConfig poolConfig = createJedisPoolConfig();
        final String[] hostAndPortSplit = connectionString.split("[:]");
        final String host = hostAndPortSplit[0].trim();
        final int port = Integer.parseInt(hostAndPortSplit[1].trim());
        return new JedisPool(poolConfig, host, port, timeout, password, dbIndex);
    }

    private JedisPoolConfig createJedisPoolConfig() {
        final JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(context.getProperty(POOL_MAX_TOTAL).asInteger());
        poolConfig.setMaxIdle(context.getProperty(POOL_MAX_IDLE).asInteger());
        poolConfig.setMinIdle(context.getProperty(POOL_MIN_IDLE).asInteger());
        poolConfig.setBlockWhenExhausted(context.getProperty(POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
        poolConfig.setMaxWaitMillis(context.getProperty(POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
        poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
        poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
        poolConfig.setNumTestsPerEvictionRun(context.getProperty(POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
        poolConfig.setTestOnCreate(context.getProperty(POOL_TEST_ON_CREATE).asBoolean());
        poolConfig.setTestOnBorrow(context.getProperty(POOL_TEST_ON_BORROW).asBoolean());
        poolConfig.setTestOnReturn(context.getProperty(POOL_TEST_ON_RETURN).asBoolean());
        poolConfig.setTestWhileIdle(context.getProperty(POOL_TEST_WHILE_IDLE).asBoolean());
        return poolConfig;
    }


    public static final long DEFAULT_TIME = 30000;

    @Override
    public boolean addLock(String key, String value, long time) {
        if (time < 0) {
            time = DEFAULT_TIME;
        }
        Jedis jedis = getJedisPool().getResource();
        String result = jedis.set(key, value, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, time);
        jedis.close();
        return LOCK_SUCCESS.equals(result);
    }

    @Override
    public boolean releaseLock(String key, String value) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Jedis jedis = getJedisPool().getResource();
        Object result = jedis.eval(script, Collections.singletonList(key), Collections.singletonList(value));
        jedis.close();
        return RELEASE_SUCCESS.equals(result);
    }

    private static boolean isInteger(final String number) {
        try {
            Integer.parseInt(number);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

}
